package com.doit.doitdata.rule

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.SparkSession
import redis.clients.jedis.Jedis

/**
  * Created by hunter.coder 涛哥  
  * 2019/4/8 17:07
  * 交流qq:657270652
  * Version: 1.0
  * 更多学习资料：https://blog.csdn.net/coderblack/
  * Description: 商圈地理位置信息知识库构建程序
  **/
object BizRuleConstructor {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName(BizRuleConstructor.getClass.getSimpleName)
      .getOrCreate()

    import spark.implicits._


    // 加载数据：
    val ds1 = spark.read.textFile("G:\\sharkdata\\bizdict")


    // 切割数据，抽取字段
    val ds2 = ds1.map(line=>{
      // ('1', '北京市', '北京市', '东城区', '王府井', '116.412987', '39.908416');
      val fields = line.split("'")

      val province = fields(3)
      val city = fields(5)
      val district = fields(7)
      val biz = fields(9)

      val lng = fields(11).toDouble
      val lat = fields(13).toDouble

      (province,city,district,biz,lng,lat)
    })


    // 用foreachPartition来对RDD中的每个分区进行处理
    ds2.foreachPartition(part=>{

      // 建redis连接
      val jedis1 = new Jedis("c701",6379)
      jedis1.select(1)  // 省市区库

      val jedis2 = new Jedis("c701",6379)
      jedis2.select(2) // 商圈名称库

      // 对分区中的每一条数据，加工后写入redis
      part.foreach(tp=>{
        val geocode = GeoHash.withCharacterPrecision(tp._6,tp._5,6).toBase32

        jedis2.set(geocode,tp._4)
        jedis1.set(geocode.substring(0,geocode.length-1),Array(tp._1,tp._2,tp._3).mkString(","))

      })

      jedis1.close()
      jedis2.close()

    })

    spark.close()
  }


}
